package com.sxu.douban.mapreduce;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Step3 {
    public static class movie {
        private String movie_id;
        private Double score;

        public movie(Double a, String b) {
            this.movie_id = b;
            this.score = a;
        }

        public String getMovie_id() {
            return movie_id;
        }

        public void setMovie_id(String movie_id) {
            this.movie_id = movie_id;
        }

        public Double getScore() {
            return score;
        }

        public void setScore(Double score) {
            this.score = score;
        }
    }

    public static class mymapper extends Mapper<Object, Text, Text, Text> {
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString().split("\\s+")[1];
            String[] score = line.substring(0, line.length() - 1).split(";");
            int len = score.length;

            // 皮尓迅
//			if (len > 3) {
//				int sum1 = 0, sum2 = 0, sum1Sq = 0, sum2Sq = 0, psum = 0;
//				for (int i = 0; i < len; i++) {
//					sum1 += Integer.parseInt(score[i].split(",")[0]);
//					sum2 += Integer.parseInt(score[i].split(",")[1]);
//					sum1Sq += Integer.parseInt(score[i].split(",")[0]) * Integer.parseInt(score[i].split(",")[0]);
//					sum2Sq += Integer.parseInt(score[i].split(",")[1]) * Integer.parseInt(score[i].split(",")[1]);
//					psum += Integer.parseInt(score[i].split(",")[0]) * Integer.parseInt(score[i].split(",")[1]);
//				}
//				double num = psum - (sum1 * sum2 / (len*1.0));
//				double den = Math.sqrt((sum1Sq - Math.pow(sum1, 2) / (len*1.0)) * (sum2Sq - Math.pow(sum2, 2) / (len*1.0)));
//
//				String MA = value.toString().split("\\s+")[0].split(",")[0];
//				String MB = value.toString().split("\\s+")[0].split(",")[1];
//				if (den == 0) {
//					context.write(new Text(MA), new Text("0," + MB));
//					context.write(new Text(MB), new Text("0," + MA));
//				} else {
//					context.write(new Text(MA), new Text(num / den + "," + MB));
//					context.write(new Text(MB), new Text(num / den + "," + MA));
//				}
//			}

            // 歐幾里得距離
            String MA = value.toString().split("\\s+")[0].split(",")[0].split("@")[0];
            String MB = value.toString().split("\\s+")[0].split(",")[1].split("@")[0];

            String MA_type=value.toString().split("\\s+")[0].split(",")[0].split("@")[1];
            String MB_type=value.toString().split("\\s+")[0].split(",")[1].split("@")[1];

            String MA_type_S[] =MA_type.split("/");
            String MB_type_S[] =MB_type.split("/");

            int len_A=MA_type_S.length;
            int len_B=MB_type_S.length;

            double flag=0;
            for(int i=0;i<len_A;i++){
                for(int j=0;j<len_B;j++){
                    if(MA_type_S[i].equals(MB_type_S[j])){
                        flag+=0.1;
                    }
                }
            }
            if (len <= 3) {

            } else {
                int sum = 0;
                for (int i = 0; i < len; i++) {
                    sum += (Integer.parseInt(score[i].split(",")[0])*10 - Integer.parseInt(score[i].split(",")[1])*10)
                            * (Integer.parseInt(score[i].split(",")[0])*10 - Integer.parseInt(score[i].split(",")[1])*10);
                }
                double sim=1.0 / (1.0 + Math.sqrt(sum * 1.0));
                sim+=flag;
                context.write(new Text(MA), new Text(String.valueOf(sim) + "," + MB));
                context.write(new Text(MB), new Text(String.valueOf(sim) + "," + MA));

            }
        }
    }

    public static class myreducer extends Reducer<Text, Text, Text, Text> {
        public void reduce(Text key, Iterable<Text> values, Context context) throws InterruptedException, IOException {
            List<movie> temp = new ArrayList<movie>();
            for (Text value : values) {
                Double a = Double.parseDouble(value.toString().split(",")[0]);
                String b = value.toString().split(",")[1];
                temp.add(new movie(a, b));
            }
            System.setProperty("java.util.Arrays.useLegacyMergeSort", "true");
           Collections.sort(temp, new Comparator<movie>() {
               @Override
               public int compare(movie o1, movie o2) {
                   return o2.getScore().compareTo(o1.getScore());
               }
           });

            int i = 0;
              String temp1 = "";
            if (temp.size() < 10) {
                for (movie mv : temp) {
                    if(++i==temp.size()){
                        temp1 += mv.getMovie_id();
                    }
                    else
                    {
                        temp1 += mv.getMovie_id()  + ",";
                    }

                }
            } else {
                for (movie mv : temp) {
                    if(++i==10){
                        temp1 += mv.getMovie_id() ;
                        break;
                    }
                    else{
                        temp1 += mv.getMovie_id()  + ",";
                    }
                }
            }
            context.write(key, new Text(temp1));
        }
    }

    public static void run() throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS","hdfs://192.168.22.131:9000");
        FileSystem fileSystem = FileSystem.get(configuration);
        Job job = Job.getInstance(configuration);
        job.setJarByClass(Step3.class);
        job.setMapperClass(Step3.mymapper.class);
        job.setReducerClass(Step3.myreducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path("/Movie_Recommend/output2"));
        Path outPath = new Path("/Movie_Recommend/output3");
        if (fileSystem.exists(outPath)){
            fileSystem.delete(outPath,true);
        }
        FileOutputFormat.setOutputPath(job, outPath);
        job.waitForCompletion(true);
    }
}
